//
//  FilterObservableViewController.swift
//  Example
//
//  Created by T AO on 2022/2/10.
//

import UIKit
import RxSwift

class FilterObservableViewController: BaseViewController {

    override func viewDidLoad() {
        super.viewDidLoad()

//        filter()
//        distinctUntilChanged()
//        elementAt()
//        single()
        
//        take()
//        takeLast()
//        takeWhile()
//        takeUntil()
        
//        skip()
//        skipUntil()
        
//        sample()
        debounce()
    }

}

// MARK: - 过滤条件操作符

extension FilterObservableViewController {
    /// filter
    /// 过滤某些不符合要求的事件
    private func filter() {
        Observable.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 0)
            .filter { $0 % 2 == 0 }
            .subscribe(onNext: { print($0) })
            .disposed(by: disposeBag)
    }
    
    /// distinctUntilChanged
    /// 过滤掉连续重复的事件
    private func distinctUntilChanged() {
        Observable.of(1, 2, 2, 2, 2, 3, 3, 3, 4, 4)
            .distinctUntilChanged()
            .subscribe(onNext: { print($0) })
            .disposed(by: disposeBag)
    }
    
    /// elementAt
    /// 只处理在指定位置的事件
    private func elementAt() {
        Observable.of("G", "e", "r", "m", "t", "a", "o")
            .element(at: 3)
            .subscribe(onNext: { print($0) })
            .disposed(by: disposeBag)
    }
    
    /// single
    /// 限制只发送一次事件，或者满足条件的第一个事件
    /// 如果存在有多个事件或者没有事件都会发出一个 error 事件
    /// 如果只有一个事件，则不会发出 error 事件
    private func single() {
        Observable.of(1, 2, 3, 4)
//            .single() // 1 Unhandled error happened: Sequence contains more than one element.
            .single { $0 == 3 } // 3
            .subscribe(onNext: { print($0) })
            .disposed(by: disposeBag)
    }
    
    /// take
    /// 仅发送 Observable 序列中的前 n 个事件，在满足数量之后会自动 .completed
    private func take() {
        Observable.of("1", "2", "3", "4")
            .take(2)
            .subscribe(onNext: { print($0) })
            .disposed(by: disposeBag)
    }
    
    /// takeLast
    /// 仅发送 Observable 序列中的后 n 个事件
    private func takeLast() {
        Observable.of("1", "2", "3", "4")
            .takeLast(2)
            .subscribe(onNext: { print($0) })
            .disposed(by: disposeBag)
    }
    
    /// takeWhile
    /// 仅发送 Observable 序列中满足条件的事件
    private func takeWhile() {
        Observable.of(1, 2, 3, 4, 5)
            .take(while: { $0 < 3 })
            .subscribe(onNext: { print($0) })
            .disposed(by: disposeBag)
    }
    
    /// takeUntil
    private func takeUntil() {
        let sourceSequence = PublishSubject<String>()
        let referenceSequence = PublishSubject<String>()
        
        sourceSequence
            .take(until: referenceSequence)
            .subscribe(onNext: { print($0) })
            .disposed(by: disposeBag)
        
        sourceSequence.onNext("Germ")
        sourceSequence.onNext("TAO")
        sourceSequence.onNext("zzz")
        
        referenceSequence.onNext("---") // 条件一出来，下面就不走了
        
        sourceSequence.onNext("sss")
        sourceSequence.onNext("ccc")
        sourceSequence.onNext("xxx")
    }
    
    /// skip
    /// 跳过源 Observable 序列发出的前 n 个事件
    private func skip() {
        Observable.of(1, 2, 3, 4, 5, 6)
//            .skip(3)
            .skip(while: { $0 < 4 })
            .subscribe(onNext: { print($0) })
            .disposed(by: disposeBag)
    }
    
    private func skipUntil() {
        let sourceSequence = PublishSubject<String>()
        let referenceSequence = PublishSubject<String>()
        
        sourceSequence
            .skip(until: referenceSequence)
            .subscribe(onNext: { print($0) })
            .disposed(by: disposeBag)
        
        // 没有条件命令，下面走不了
        sourceSequence.onNext("Germ")
        sourceSequence.onNext("TAO")
        sourceSequence.onNext("zzz")
        
        referenceSequence.onNext("---") // 条件一出来，下面就可以走了
        
        sourceSequence.onNext("sss")
        sourceSequence.onNext("ccc")
        sourceSequence.onNext("xxx")
    }
    
    /// sample
    /// 除了订阅源 Observable 外，还可以监视另外一个 Observable， 即 notifier
    /// 每当收到 notifier 事件，就会从源序列取一个最新的事件并发送。而如果两次 notifier 事件之间没有源序列的事件，则不发送值
    private func sample() {
        let source = PublishSubject<Int>()
        let notifier = PublishSubject<String>()
        
        source
            .sample(notifier)
            .subscribe(onNext: { print($0) })
            .disposed(by: disposeBag)
        
        source.onNext(1)
        
        // 让源序列接收消息
        notifier.onNext("A") // 打印：1
        
        source.onNext(2)
        
        // 让源序列接收消息
        notifier.onNext("B")
        notifier.onNext("C")
        
        source.onNext(3)
        source.onNext(4)
        
        // 让源序列接收消息
        notifier.onNext("D")
        
        source.onNext(5)
        
        // 让源序列接收消息
        notifier.onCompleted()
    }
    
    /// debounce
    /// 过滤掉高频产生的元素，它只会发出这种元素：该元素产生后，一段时间内没有新元素产生
    /// 换句话说就是，队列中的元素如果和下一个元素的间隔小于了指定的时间间隔，那么这个元素将被过滤掉
    /// 常用在用户输入的时候，不需要每个字母敲进去都发送一个事件，而是稍等一下取最后一个事件
    private func debounce() {
        // 定义好每个事件里的值以及发送的时间
        let times = [
            ["value": 1, "time": 0.1],
            ["value": 2, "time": 1.1],
            ["value": 3, "time": 1.2],
            ["value": 4, "time": 1.2],
            ["value": 5, "time": 1.4],
            ["value": 6, "time": 2.1],
        ]
        
        // 生成对应的 Observable 序列并订阅
        Observable.from(times)
            .flatMap { item in
                return Observable.of(Int(item["value"]!))
                    .delaySubscription(RxTimeInterval.seconds(Int(item["time"]!)), scheduler: MainScheduler.instance)
            }
            .debounce(RxTimeInterval.milliseconds(500), scheduler: MainScheduler.instance) // 只发出与下一个间隔超过 0.5 秒的元素
            .subscribe(onNext: { print($0) })
            .disposed(by: disposeBag)
    }
}
